package hasthi.container.p2p;

import hasthi.common.constants.HasthiConstants;
import hasthi.manager.BroadcastDocument;
import hasthi.manager.BroadcastDocument.Broadcast;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.w3.x2005.x08.addressing.ToDocument;

import rice.pastry.Id;
import rice.pastry.NodeHandle;
import rice.pastry.Id.Distance;
import rice.pastry.leafset.LeafSet;
import rice.pastry.routing.RouteSet;
import rice.pastry.routing.RoutingTable;

public class Broadcaster {
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);

    public static final Distance L_DIST = new Distance(Id.One.clone()).shift(-16, 0);
    public static final Distance NEG_L_BY2 = new Distance(Id.NegOne.clone()).shift(-15, 0);
    public static final Distance POS_L_BY2 = new Distance(Id.One.clone()).shift(-15, 0);

    public static final rice.pastry.Id ID_ONE = rice.pastry.Id.build(rice.pastry.Id.One.clone());
    public static final rice.pastry.Id START_ID = ID_ONE.getCCW();
    public static final rice.pastry.Id MIDDLE_ID = rice.pastry.Id.build(new Distance(
            rice.pastry.Id.One.clone()).shift(-1 * (rice.pastry.Id.IdBitLength - 1), 0).copy());
    public static final rice.pastry.Id END_ID = START_ID.getCCW();

    private P2PContainer container;
    private Map<String, String> recivedMessageMap = new Hashtable<String, String>();

    public Broadcaster(P2PContainer container) {
        this.container = container;
    }

    public boolean broadcast(UnicastXmlMessage message) {
        return broadcast(message, true);
    }

    public boolean broadcast(UnicastXmlMessage message, boolean dropduplicate) {
        if (dropduplicate && recivedMessageMap.containsKey(message.getMessageID())) {
            return true;
        }

        recivedMessageMap.put(message.getMessageID(), message.getMessageID());
        Broadcast broadcast = (Broadcast) message
                .decendentByName(HasthiConstants.BROADCAST_HEADER);
        broadcast.setTreedepth(broadcast.getTreedepth() + 1);

        Map<String, String> map = new HashMap<String, String>();
        if (broadcast.getReceivedArray() != null) {
            for (String id : broadcast.getReceivedArray()) {
                map.put(id, id);
            }
        }

        if (!map.containsKey(container.getNode().getId())) {
            broadcast.addNewReceived().setStringValue(container.getNode().getId().toStringFull());
        }

        Id start = Id.build(broadcast.getStartAddress());
        Id end = Id.build(broadcast.getEndAddress());

        RoutingTable rt = container.getNode().getRoutingTable();

        Id myId = container.getNode().getNodeId();

        Id mid1;
        Id mid2;

        // find the neighbour range
        if (myId.isBetween(start, end)) {
            mid1 = myId.add(NEG_L_BY2);
            mid2 = myId.add(POS_L_BY2);
        } else {
            mid1 = mean(start, end);
            mid2 = mid1;
        }
        /**
         * adjust mid1 and mid2 s.t. mid1 = Max(start, myid - L/2) and mid2 =
         * Min(end, myid + L/2) where L = 2^16
         */
        if (start.isBetween(mid1, mid2)) {
            start = mid1;
        }
        if (end.isBetween(mid1, mid2)) {
            end = mid2;
        }

        // find all the entires between start and end
        SortedSet<Id> outgoingList1 = new TreeSet<Id>();
        SortedSet<Id> outgoingList2 = new TreeSet<Id>();
        Set<Id> neighbourList = new HashSet<Id>();

        for (int i = 0; i < rt.numRows(); i++) {
            RouteSet[] rows = rt.getRow(i);
            for (RouteSet set : rows) {
                if (set != null && set.closestNode() != null) {
                    for (int j = 0; j < set.size(); j++) {
                        Id id = set.get(j).getNodeId();
                        if(myId.equals(id)){
                            continue;
                        }
                        if (id.isBetween(mid1, mid2)) {
                            neighbourList.add(id);
                        } else if (id.isBetween(start, mid1)) {
                            outgoingList1.add(id);
                        } else if (id.isBetween(mid2, end)) {
                            outgoingList2.add(id);
                        }
                    }
                }
            }
        }

        LeafSet nodeSet = container.getNode().getLeafSet();

        for (NodeHandle node : (List<NodeHandle>) nodeSet.asList()) {
            Id id = node.getNodeId();
            if (!map.containsKey(id.toStringFull()) && !myId.equals(id)) {
                if (id.isBetween(mid1, mid2)) {
                    neighbourList.add(id);
                } else if (id.isBetween(start, mid1)) {
                    outgoingList1.add(id);
                } else if (id.isBetween(mid2, end)) {
                    outgoingList2.add(id);
                }else{
                    System.out.println("[" + start + "," + end + "] not in" + id);    
                }
            } 
        }

        int recipentCount = 0;
        if (mid1 != mid2) {
            recipentCount = sendToLeafSet(message, broadcast, neighbourList);
        }
        sendToTree(outgoingList1, message, broadcast, start, mid1);
        sendToTree(outgoingList2, message, broadcast, mid2, end);
        recipentCount = recipentCount + outgoingList1.size() + outgoingList2.size();
        log.debug("[(" + broadcast.getTreedepth() + ")" + container.getNode().getId()
                + "]broadcast [" + start + "," + end + "] "
                + recipentCount);
        log.debug("broadcast Size =" + neighbourList.size());
        return false;
    }

    public void initiateBroadcast(UnicastXmlMessage multiCastMessage){
        ToDocument toDocument = ToDocument.Factory.newInstance();
        toDocument.addNewTo().setStringValue(HasthiConstants.BROADCAST_ADDRESS);
        multiCastMessage.addHeader(toDocument);
        
        BroadcastDocument broadcastDocument = BroadcastDocument.Factory.newInstance();
        Broadcast broadcast = broadcastDocument.addNewBroadcast();
        broadcast.setIstreeMode(true);
        multiCastMessage.addHeader(broadcastDocument);
        
        broadcast = multiCastMessage.getBroadcast();
        broadcast.setStartAddress(Broadcaster.START_ID.toStringFull());
        broadcast.setEndAddress(Broadcaster.MIDDLE_ID.toStringFull());
       
        broadcast(multiCastMessage,false);
        
        broadcast.setStartAddress(Broadcaster.MIDDLE_ID.toStringFull());
        broadcast.setEndAddress(Broadcaster.END_ID.toStringFull());
        broadcast.setTreedepth(-1);
        broadcast(multiCastMessage,false);
    }
    
    // public boolean broadcast(XmlMessage message, boolean dropduplicate) {
    // if(dropduplicate &&
    // recivedMessageMap.containsKey(message.getMessageID())){
    // return true;
    // }
    // recivedMessageMap.put(message.getMessageID(), message.getMessageID());
    // Broadcast broadcast
    // = (Broadcast) message.decendentByName(ManagerConstants.BROADCAST_HEADER);
    // broadcast.setTreedepth(broadcast.getTreedepth()+1);
    //        
    // Id start = Id.build(broadcast.getStartAddress());
    // Id end = Id.build(broadcast.getEndAddress());
    //        
    //        
    //
    // RoutingTable rt = container.getNode().getRoutingTable();
    //
    // Distance rangeSize = start.distance(end);
    // if (!broadcast.getIstreeMode() || rangeSize.compareTo(L_DIST) < 0) {
    // int recipentCount = sendToLeafSet(message, broadcast, start, end);
    // log.debug("[("+ broadcast.getTreedepth() +")" +
    // container.getNode().getId() + "]broadcast ["+ start.toStringFull() + ","
    // + end.toStringFull() + "] "+ recipentCount);
    // } else {
    // Id myId = container.getNode().getNodeId();
    //            
    // Id mid1; Id mid2;
    //            
    // if(myId.isBetween(start, end)){
    // mid1 = myId.add(NEG_L_BY2);
    // mid2 = myId.add(POS_L_BY2);
    // }else{
    // mid1 = mean(start, end);
    // mid2 = mid1;
    // }
    // // find all the entires between start and end
    // SortedSet<Id> outgoingList1 = new TreeSet<Id>();
    // SortedSet<Id> outgoingList2 = new TreeSet<Id>();
    // for (int i = 0; i < rt.numRows(); i++) {
    // RouteSet[] rows = rt.getRow(i);
    // for (RouteSet set : rows) {
    // if (set != null && set.closestNode() != null) {
    // for (int j = 0; j < set.size(); j++) {
    // Id id = set.get(j).getNodeId();
    // if (id.isBetween(start, mid1)) {
    // outgoingList1.add(id);
    // } else if (id.isBetween(mid2, end)) {
    // outgoingList2.add(id);
    // }
    // }
    // }
    // }
    // }
    // int recipentCount = 0;
    // if(mid1 != mid2){
    // recipentCount = sendToLeafSet(message, broadcast, mid1, mid2);
    // }
    // sendToTree(outgoingList1, message, broadcast, start, mid1);
    // sendToTree(outgoingList2, message, broadcast, mid2, end);
    // recipentCount = recipentCount + outgoingList1.size() +
    // outgoingList2.size();
    // log.debug("[("+ broadcast.getTreedepth() +")"+
    // container.getNode().getId() + "]broadcast ["+ start.toStringFull() + ","
    // + end.toStringFull() + "] "+ recipentCount);
    // }
    // return false;
    // }

    public void sendToTree(SortedSet<Id> outgoingList, UnicastXmlMessage message, Broadcast broadcast,
            Id start, Id end) {
        Id startr = start;
        Id endr = null;
        Id lastid = null;
        for (Id id : outgoingList) {
            if (lastid == null) {
                lastid = id;
            } else {
                endr = mean(lastid, id);
                if (lastid.distance(endr).compareTo(L_DIST) >= 0) {
                    broadcast.setStartAddress(startr.toStringFull());
                    broadcast.setEndAddress(endr.toStringFull());
                    container.sendMessage(lastid, message);
                    startr = endr;
                }
                lastid = id;
            }
        }
        if (lastid != null) {
            broadcast.setStartAddress(startr.toStringFull());
            broadcast.setEndAddress(end.toStringFull());
            container.sendMessage(lastid, message);
        }
    }

    public Id mean(Id id1, Id id2) {
        return id1.add(id1.distance(id2).shift(1, 0));
    }

    private int sendToLeafSet(UnicastXmlMessage messsage, Broadcast broadcast, Set<Id> outgoingList) {
        broadcast.setIstreeMode(false);
        for (Id id : outgoingList) {
            broadcast.addReceived(id.toStringFull());
        }

        for (Id id : outgoingList) {
            container.sendMessage(id, messsage);
        }
        return outgoingList.size();
    }

    // private int sendToLeafSet(XmlMessage messsage, Broadcast broadcast, Id
    // start, Id end) {
    // Map<String, String> map = new HashMap<String, String>();
    // if (broadcast.getReceivedArray() != null) {
    // for (String id : broadcast.getReceivedArray()) {
    // map.put(id, id);
    // }
    // }
    //
    // if (!map.containsKey(container.getNode().getId())) {
    // broadcast.addNewReceived().setStringValue(container.getNode().getId().toStringFull());
    // }
    //
    //        
    //        
    // LeafSet nodeSet = container.getNode().getLeafSet();
    //        
    // Set<Id> outgoingList = new HashSet<Id>();
    // for (NodeHandle node : (List<NodeHandle>) nodeSet.asList()) {
    // Id id = node.getNodeId();
    // if (id.isBetween(start, end)
    // && !map.containsKey(id.toStringFull())
    // && !container.getNode().getId().equals(id)) {
    // outgoingList.add(id);
    // }else{
    // System.out.println("[" + start +"," + end + "] not in" + id);
    // }
    // }
    //
    // broadcast.setIstreeMode(false);
    // for (Id id : outgoingList) {
    // broadcast.addReceived(id.toStringFull());
    // }
    //
    // for (Id id : outgoingList) {
    // container.sendMessage(id, messsage);
    // }
    // return outgoingList.size();
    // }
}
